package sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

/**

 使用watermark 过滤迟到的数据 可以使用
    WHERE CURRENT_WATERMARK(rowtime) is null OR rowtime >= CURRENT_WATERMARK(rowtime)

 任何小于当前最大时间的数据都将被忽略。

 但是如果需要针对sn进行聚合，这种办法就不行了

 

 */
public class    D8_KafakSource_current_watermark {


    public static void main(String[] args) {

        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // String genSql = "CREATE TABLE ods_tb ( " +
        //     " stime STRING," +
        //     " name STRING," +
        //     " id BIGINT," +
        //     " rowtime AS to_timestamp(stime)," +
        //     " WATERMARK FOR rowtime AS rowtime - interval '0' second" +
        //     ") WITH ( " +
        //     "  'connector' = 'kafka'," +
        //     "  'topic' = 'test'," +
        //     "  'properties.bootstrap.servers' = 'wsl:9092'," +
        //     "  'properties.group.id' = 'testGroup'," +
        //     "  'scan.startup.mode' = 'latest-offset'," +
        //     "  'format' = 'json'" +
        //     ")";


        String genSql = "CREATE TABLE ods_tb ( " +
            " stime TIMESTAMP(3)," +
            " name STRING," +
            " id BIGINT," +
            " rowtime AS localtimestamp," +
            " WATERMARK FOR rowtime AS rowtime - interval '0' second" +
            ") WITH ( " +
            " 'connector'='datagen', " +
            " 'rows-per-second'='1'," +
            " 'fields.name.length'='1'," +
            " 'fields.id.kind'='sequence'," +
            " 'fields.id.start'='1'," +
            " 'fields.id.end'='10000000'" +
            ")";


        String print = "CREATE TABLE print (" +"    " +
            "    stime TIMESTAMP(3), " +
            "    name STRING, " +
            "    id BIGINT" +
            ") WITH (" +
            "     'connector' = 'print'" +
            ")";

        /**
         *
         */
        String sql = "insert into print " +
            " SELECT " +
            " stime," +
            " name," +
            " id " +
            " FROM ods_tb " +
            " WHERE CURRENT_WATERMARK(rowtime) is null OR rowtime >= CURRENT_WATERMARK(rowtime)";



        tableEnv.executeSql(genSql);
        tableEnv.executeSql(print);

        System.out.println(sql);
        tableEnv.executeSql(sql);


        System.out.println(new Timestamp(System.currentTimeMillis()));


    }
}
